/*********************************************************************************
 *      Copyright:  (C) 2021 litupei<lismilepei@163.com>
 *                  All rights reserved.
 *
 *       Filename:  prj_mosquitto_thread.h
 *    Description:  This file is prj_mosquitto_thread.h
 *                 
 *        Version:  1.0.0(2021年08月10日)
 *         Author:  litupei <lismilepei@163.com>
 *      ChangeLog:  1, Release initial version on "2021年08月10日 17时23分47秒"
 *                 
 ********************************************************************************/

#include "prj_mosquitto_thread.h"

#if 1

int thread_start(pthread_t *tid, THREAD_BODY *thread_workfunc, void *thread_arg)
{
    int             rv = -1;
    pthread_attr_t  thread_attr;
    
    if ( !tid || !thread_workfunc || !thread_arg )
    {
        printf ("Invalid parameter in %s\n\n", __FUNCTION__);
        return rv;
    }
    
    //初始化线程属性
    if (pthread_attr_init(&thread_attr))
    {
        printf("pthread_attr_init() failure: %s\n", strerror(errno));
        pthread_attr_destroy(&thread_attr);
        return rv;
    }
    
    //设置栈大小
    if (pthread_attr_setstacksize(&thread_attr, 120*1024))
    {
        printf("pthread_attr_setstacksize() failure: %s\n", strerror(errno));
        pthread_attr_destroy(&thread_attr);
        return rv;
    }
    
    //设置为相分离，主线程就不需要在子线程退出的时候与其会合
    if (pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED))
    {
        printf("pthread_attr_setdetachstate() failure: %s\n", strerror(errno));
        pthread_attr_destroy(&thread_attr);
        return rv;
    }
    
    //创建一个子线程
    if (pthread_create(tid, &thread_attr, thread_workfunc, thread_arg))
    {
        printf("pthread_create failure: %s\n", strerror(errno));
        pthread_attr_destroy(&thread_attr);
        return rv;
    }
 
    rv = 0;
    pthread_attr_destroy(&thread_attr);
    //线程属性在使用完之后，调用pthread_attr_destroy把它摧毁释放
    return rv;
 
}

#endif

void *pub_connect(void *args)
{
	int		rv = 0;
	mqtt_cxt_t	*mqtt = 	NULL;
	struct mosquitto	*mosq = NULL;

	mqtt = (mqtt_cxt_t *)args;
   	if(sqlite3_open(FILENAME, &mqtt->db) != SQLITE_OK)
	{
		printf("open the sqlite is failture : %s \n", sqlite3_errmsg(mqtt->db));
		sqlite3_close(mqtt->db);
		return NULL;
	}
	printf("openning the sqlite3 database named student.db successfully! \n");

#if 1
	if( !create_table(mqtt->db) )
	{
		printf("the Sensor table is exit!\n");
	}
	else
	printf("the Sensor table create successfully! \n");
#endif

#if 1
    printf("**********publish message************* \n");
	printf("%s \n", mqtt->username);
	printf("%s \n", mqtt->pub_password);
	printf("%s \n", mqtt->host);
	printf("%s \n", mqtt->pub_client);
	printf("%d \n", mqtt->port);
	printf("%d \n", mqtt->keepalive);
	printf("%s \n", mqtt->pub_topic);
    printf("************************************** \n");
#endif

	mosq = mosquitto_new(mqtt->pub_client, true, (void *)mqtt);

	if( !mosq )
	{
		printf("Create a new pub_mosquitto client instance failture! :%s \n", strerror(errno));
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}
	printf("Create a new pub_mosquitto client instance successfully! \n");
	
	//回调函数	
	mosquitto_connect_callback_set(mosq, pub_connect_callback);
   	mosquitto_disconnect_callback_set(mosq, pub_on_disconnect);
    	mosquitto_publish_callback_set(mosq, pub_on_publish);

	if( mosquitto_username_pw_set(mosq, mqtt->username, mqtt->pub_password) !=MOSQ_ERR_SUCCESS )	
	{
   		printf("pub_mosquitto username and passwd failure:%s\n",strerror(errno));
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		return NULL;
	}
	printf("pub_mosquitto username and password successfully! \n");

	while( 1 )
	{
        	found_max_id(mqtt->db, &mqtt->sq_id);
        	printf("sq_id : %d \n", mqtt->sq_id);
	        if( ds18b20_temp(&mqtt->temp) < 0 )
        	{
            		printf("ds18b20 get temp failure\n");
            		return NULL;
        	}
 
	        mqtt->timenow = time_get();
		printf("%s \n", mqtt->timenow);
        	insert_data(mqtt->db, mqtt->timenow, mqtt->temp, mqtt->sq_id);

        	printf("starting publish connect..... \n");
		if( mosquitto_connect(mosq, mqtt->host, mqtt->port, mqtt->keepalive) != MOSQ_ERR_SUCCESS )
		{			
			printf("pub_mosquitto connect server failure:%s\n",strerror(errno));
			sleep(1);
			continue;		
		}
		printf("pub_mosquitto connect server successfully! \n");
	#if 0
		if( mosquitto_loop_forever(mosq, -1, 1 ) != MOSQ_ERR_SUCCESS )
		{
			printf("pub_mosquitto loop failure:%s\n",strerror(errno));
			mosquitto_destroy(mosq);
			mosquitto_lib_cleanup();
			return NULL;
		}
		printf("pub_mosquitto loop successfully! \n");
	#endif
        	mosquitto_loop_forever(mosq, -1, 1 );
	        printf("starting sleep 5s...... \n");
		sleep(5);
	        //printf("starting publish connect..... \n");
	}

	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();

	return NULL;
}

void pub_connect_callback(struct mosquitto *mosq, void *obj, int rv)
{
	mqtt_cxt_t	*mqtt = NULL;

	if(rv)
	{
		printf("publish : %s \n", mosquitto_connack_string(rv));
		return;
	}
	printf("publish : %s \n", mosquitto_connack_string(rv));
    
	//memset(mqtt, 0, sizeof(mqtt));
	mqtt = (mqtt_cxt_t *)obj;

    	if(mqtt == NULL)
    	{
        	printf("the obj is null ! \n");
        	mosquitto_disconnect(mosq);
        	return;
    	}

	mqtt->pub_payload = pack_json(mqtt->temp);	    
	printf("%s \n",mqtt->pub_payload);
	printf("pub_topic : %s \n",mqtt->pub_topic);

	if( mosquitto_publish(mosq, &mqtt->pub_mid, mqtt->pub_topic, strlen(mqtt->pub_payload), mqtt->pub_payload, 0 , false) != MOSQ_ERR_SUCCESS )  
	{   
		printf("publish error !: %s\n", strerror(errno));
		free(mqtt->pub_payload);
		mosquitto_disconnect(mosq);
		return;
	}
    	printf("publish successfully! \n");

	free(mqtt->pub_payload);
	mosquitto_disconnect(mosq);
	
}

void *sub_connect(void *args)
{
	int		        rv = 0;
	mqtt_cxt_t	        *mqtt = NULL;
	struct mosquitto	*mosq = NULL;

	mqtt = (mqtt_cxt_t *)args;
	//ini_mqttmsg(&mqtt);

#if 1
   	printf("***********subscribe message************** \n");
	printf("%s \n", mqtt->username);
	printf("%s \n", mqtt->sub_password);
	printf("%s \n", mqtt->host);
	printf("%s \n", mqtt->sub_client);
	printf("%d \n", mqtt->port);
	printf("%d \n", mqtt->keepalive);
	printf("%s \n", mqtt->sub_topic);
    	printf("****************************************** \n");
#endif

	mosq = mosquitto_new(mqtt->sub_client, true, (void *)mqtt);

	if( !mosq )
	{
		printf("Create a new sub_mosquitto client instance failture! :%s \n", strerror(errno));
		mosquitto_destroy(mosq);
		//mosquitto_lib_cleanup();
		return NULL;
	}
	printf("Create a new sub_mosquitto client instance successfully! \n");
	
	mosquitto_connect_callback_set(mosq, sub_connect_callback);
	mosquitto_disconnect_callback_set(mosq, sub_disconnect_callback);
    	mosquitto_subscribe_callback_set(mosq, sub_subscribe_callback);
    	mosquitto_message_callback_set(mosq, sub_message);

	if( mosquitto_username_pw_set(mosq, mqtt->username, mqtt->sub_password) !=MOSQ_ERR_SUCCESS )	
	{
   		printf("sub_mosquitto username and passwd failure:%s\n",strerror(errno));
		mosquitto_destroy(mosq);
		//mosquitto_lib_cleanup();
		return NULL;
	}
	printf("sub_mosquitto username and password successfully! \n");

	while( 1 )
	{

		if( mosquitto_connect(mosq, mqtt->host, mqtt->port, mqtt->keepalive) != MOSQ_ERR_SUCCESS )
		{
			printf("sub_mosquitto connect server failure:%s\n",strerror(errno));
			sleep(1);
			continue;
		}
		printf("sub_mosquitto connect server successfully! \n");
	#if 1
		if( mosquitto_loop_forever(mosq, -1, 1 ) != MOSQ_ERR_SUCCESS )
		{
			printf("sub_mosquitto loop failure:%s\n",strerror(errno));
			mosquitto_destroy(mosq);
			//mosquitto_lib_cleanup();
			return NULL;
		}
		printf("sub_mosquitto loop successfully! \n");
	#endif

	//	sleep(10);
	}
	mosquitto_destroy(mosq);
	//mosquitto_lib_cleanup();

	return NULL;
}

void sub_connect_callback(struct mosquitto *mosq, void *obj, int rv)
{
	mqtt_cxt_t	*mqtt = NULL;

	if(rv)
	{
		printf("subscribe : %s \n", mosquitto_connack_string(rv));
		mosquitto_disconnect(mosq);
		return;
  	}
	printf("subscribe success: %s \n", mosquitto_connack_string(rv));

	//memset(mqtt, 0, sizeof(mqtt));
	mqtt = (mqtt_cxt_t *)obj;
    	if(mqtt == NULL)
    	{
        	printf("the obj is null! \n");
    	}

	if( mosquitto_subscribe(mosq, &mqtt->sub_mid, mqtt->sub_topic, 0) != MOSQ_ERR_SUCCESS )  
	{   
		printf("subscribe error !: %s\n", strerror(errno));
       		return;
	}
    	printf("subscribe successfully! \n");

}

void sub_disconnect_callback(struct mosquitto *mosq, void *obj, int rv)
{
	printf ("sub disconnected callback\n");
}

void sub_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
	printf("sub_subscribe successfully! \n");
}

void sub_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
	int         rv = 0;
	mqtt_cxt_t  *mqtt= NULL;
	int         ctrl_led;
    
	mqtt = (mqtt_cxt_t *)obj;
    
	memset(mqtt->sub_payload, 0, sizeof(mqtt->sub_payload));
	strncpy(mqtt->sub_payload, (char *)message->payload, sizeof(mqtt->sub_payload));
    
	printf ("sub_payload msg: %s\n", mqtt->sub_payload);
	printf ("payload_len: %d\n", strlen(mqtt->sub_payload));

	rv = parse_payload(mqtt->sub_payload, &ctrl_led);
	if(rv < 0)
	{
        	printf("parse_payload failure! \n");
        	return;
    	}

	if(ctrl_led == 1)
	{
        	printf("the led openning! \n");
    	}

    	else
        printf("the led closed! \n");

}

void pub_on_disconnect(struct mosquitto *mosq, void *obj, int rc)
{   
	printf ("publish callback on_disconnect\n\n");
}
 

void pub_on_publish(struct mosquitto *mosq, void *obj, int mid)
{
	mqtt_cxt_t  *mqtt = NULL;
    
	mqtt = (mqtt_cxt_t *)obj;

	delete_data(mqtt->db, mqtt->sq_id);

	printf ("callback on_publish\n");
	printf ("publish message %d has been posted to the cloud\n\n", mid);
}



